/**
@file
	MassiveConnectionDevice.cpp
@brief
	Implentation of MassiveConnectionDevice.
@author
	GengYong
@revision
	2005/10/27 * Created by GengYong.
*/

#include "ConnectionDevice.h"
using namespace MMONetwork::Server;

#include <new>
#include <memory>
#include <cassert>

//----------------------------------------------------------------------------
//	Debugger
//----------------------------------------------------------------------------
#ifdef _DEBUG
#	define LOG
#	define WARNING LOG
#	define DBGOUT  LOG
#else
#	define LOG
#	define WARNING LOG
#	define DBGOUT  LOG
#endif


//----------------------------------------------------------------------------
//	Internal Header
//----------------------------------------------------------------------------
namespace Internal
{
	typedef struct PackageHeader
	{
		unsigned short length;
		unsigned short routine;	
		unsigned short segment;
		unsigned short command;
	} PackageHeader;
}

//----------------------------------------------------------------------------
// inline tool functions
//----------------------------------------------------------------------------
// Disable/Enable address/port reuseable.
inline bool SetSocketReuseAddr(SOCKET sock, bool flag)
{
	int optval = flag ? 1:0;
#if _MSC_VER
	return (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (char *)&optval, sizeof(optval)) != 0);
#else
	return (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval)) != 0);
#endif
}

// Disable/Enable TCP Nagle algorithm. Only valid for TCP Stream.
inline bool SetSocketNoDelay(SOCKET sock, bool flag)
{
	int optval = flag ? 1:0;
#if _MSC_VER
	return (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (char *)&optval, sizeof(optval)) != 0);
#else
	return (setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, &optval, sizeof(optval)) != 0);
#endif
}


// Warning: enabling SO_LINGER with a nonzero time-out on a nonblocking socket is not recommended.
inline bool SetSocketLinger(SOCKET sock, bool flag, unsigned short timeout)
{			
	LINGER ling;
	ling.l_onoff  = flag;		// linger switch
	ling.l_linger = timeout;	// timeout in seconds		
	return (setsockopt(sock, SOL_SOCKET, SO_LINGER, (char *)&ling, sizeof(ling) ) == 0);
}


//----------------------------------------------------------------------------
//	inline member functions
//----------------------------------------------------------------------------
inline void MassiveConnectionDevice::_RaiseMessage(const ConnEntity& conn, const MessageEntity& msge, void * key)
{
	assert(this->m_MsgDescBuf != NULL);

	if (this->m_MsgDescCursor < this->m_MaxMessages)
	{
		MassiveConnectionDevice::MessageDescriptor * raisemsg_desc = &(this->m_MsgDescBuf[this->m_MsgDescCursor++]);

		raisemsg_desc->key		= key;
		raisemsg_desc->conn		= conn;
		raisemsg_desc->msge.buf	= msge.buf;
		raisemsg_desc->msge.len	= msge.len;
		raisemsg_desc->msge.cmd	= msge.cmd;
		raisemsg_desc->msge.seg	= msge.seg;
		raisemsg_desc->msge.time = msge.time;

		DBGOUT("A message raised. cmd:%d, seg:%d, len:%lu, time:%lu", 
			msge.cmd, msge.seg, msge.len, msge.time);
	}
	return;
}


inline bool MassiveConnectionDevice::_InsertNode(unsigned short group_id, unsigned long node_id, SOCKET sock, in_addr & addr, void * key, bool is_listen)
{
	assert(this->m_ConnDescUnusedCursor > 0);

	if (this->m_ConnDescUnusedCursor > 0)
	{
		MassiveConnectionDevice::ConnectionDescriptor * point = this->m_ConnDescUnused[--this->m_ConnDescUnusedCursor];

		point->sock				= sock;
		point->node				= node_id;
		point->group			= group_id;
		point->is_listen		= is_listen;
		point->key				= key;
		point->flux_active_time = 0;	//!!!!!!
		point->remote_addr		= addr;

		
		this->m_Sock2ConnMapper->replace((unsigned long)(sock), point);
		this->m_Node2ConnMapper[group_id %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->replace((unsigned long)(node_id), point);		
		this->m_RecvConnListBuf[this->m_RecvConnCursor] = point;
		this->m_Sock2RecvListMapper->replace((unsigned long)(sock), &(this->m_RecvConnListBuf[this->m_RecvConnCursor]));

		this->m_RecvConnCursor++;
		
		LOG("%s %s @Node(%d:%lu).", inet_ntoa(addr), (is_listen ? "listening":"connected"), group_id, node_id);
		
		if (!is_listen)
		{
			// Generate a new-conn-message.				
			ConnEntity		tmpConn;
			tmpConn.group	= group_id;
			tmpConn.node	= node_id;
			tmpConn.routine = 0;

			MessageEntity	tmpMsge;
			tmpMsge.cmd		= DEV_CONNECTION_CREATE;
			tmpMsge.buf		= NULL;
			tmpMsge.len		= 0;
			tmpMsge.time	= 0;	//!!!

			_RaiseMessage(tmpConn, tmpMsge, point->key);
		}
	}
	return true;
}

//----------------------------------------------------------------------------
//	Constructor & Destructor.
//----------------------------------------------------------------------------
MassiveConnectionDevice::MassiveConnectionDevice():
m_Sock2ConnMapper(NULL),
m_Sock2SendListMapper(NULL),
m_Sock2RecvListMapper(NULL),
m_MsgDescBuf(NULL),
m_ConnDescBuf(NULL),
m_ConnDescUnused(NULL),
m_SendConnListBuf(NULL),
m_RecvConnListBuf(NULL),
m_SelectBuffer(NULL),
m_SendBuffers(NULL),
m_RecvBuffers(NULL),
m_BasePointer(NULL),
m_MaxConnections(0),
m_MaxMessages(0),
m_SendBufLength(0),
m_RecvBufLength(0),
m_MsgDescCursor(0),
m_MsgProcCursor(0),
m_MsgCursor(0),
m_ConnDescUnusedCursor(0),
m_SendConnCursor(0L),
m_RecvConnCursor(0L),
m_FluxInBytes(0),
m_FluxOutBytes(0),
m_FluxInPackets(0),
m_FluxOutPackets(0),
m_FluxActiveTime(0)
{
#if defined (_WIN32)
	WSADATA wsd;
	WSAStartup(WINSOCK_VERSION, &wsd);
#endif
	for (unsigned int i = 0; i < (sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0])); i++)
	{
		m_Node2ConnMapper[i] = NULL;
	}
	return;
}

MassiveConnectionDevice::~MassiveConnectionDevice()
{
	Destroy();
#if defined(_WIN32)
	WSACleanup();
#endif
	return;
}

//----------------------------------------------------------------------------
//	Implentations.
//----------------------------------------------------------------------------
bool MassiveConnectionDevice::Build(const BuildParam & param)
{
	assert(param.MaxConnections > 0);
	assert(param.MaxMessages > 0);
	assert(param.RecvBufLength > 0);
	assert(param.SendBufLength > 0);
	
	Destroy();

	this->m_SendTimeout.tv_sec = param.SendTimeout / 1000;
	this->m_SendTimeout.tv_usec = param.SendTimeout % 1000;
	this->m_RecvTimeout.tv_sec = param.RecvTimeout / 1000;
	this->m_RecvTimeout.tv_usec = param.RecvTimeout % 1000;

	// re-alloc resource.
	unsigned long sum =	sizeof(MassiveConnectionDevice::MessageDescriptor) * param.MaxMessages +
						sizeof(MassiveConnectionDevice::ConnectionDescriptor) * param.MaxConnections +
						sizeof(MassiveConnectionDevice::ConnectionDescriptor*) * param.MaxConnections +
						sizeof(MassiveConnectionDevice::ConnectionDescriptor*) * param.MaxConnections +
						sizeof(MassiveConnectionDevice::ConnectionDescriptor*) * param.MaxConnections +
						sizeof(unsigned int) + sizeof(SOCKET) * param.MaxConnections +
						(sizeof(void*) + param.SendBufLength + sizeof(Internal::PackageHeader)) * param.MaxConnections +
						(sizeof(void*) + param.RecvBufLength + sizeof(Internal::PackageHeader)) * param.MaxConnections;
	try
	{
		DBGOUT("Memory: %lu Bytes(%f KB) allocted for MassiveConnectionDevice.", sum, sum/1024.0);

		this->m_BasePointer = new char[sum];
		for (unsigned int i = 0; i < (sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0])); i++)
		{
			this->m_Node2ConnMapper[i] = new Utility::mapper<ConnectionDescriptor*>(param.MaxConnections);
		}
		this->m_Sock2ConnMapper = new Utility::mapper<MassiveConnectionDevice::ConnectionDescriptor*>(param.MaxConnections);
		this->m_Sock2SendListMapper = new Utility::mapper<MassiveConnectionDevice::ConnectionDescriptor**>(param.MaxConnections);
		this->m_Sock2RecvListMapper = new Utility::mapper<MassiveConnectionDevice::ConnectionDescriptor**>(param.MaxConnections);

		memset(this->m_BasePointer, 0, sum);

		this->m_MaxConnections = param.MaxConnections;
		this->m_MaxMessages	   = param.MaxMessages;
		this->m_SendBufLength  = param.SendBufLength;
		this->m_RecvBufLength  = param.RecvBufLength;

		this->m_MsgDescCursor = 0;
		this->m_MsgProcCursor = 0;
		this->m_MsgCursor = 0;
		this->m_ConnDescUnusedCursor = param.MaxConnections;
		this->m_SendConnCursor = 0;
		this->m_RecvConnCursor = 0;
		
		// Build tables
		char * pointer = this->m_BasePointer;	
		this->m_MsgDescBuf = (MassiveConnectionDevice::MessageDescriptor*)pointer;
		pointer += sizeof(MassiveConnectionDevice::MessageDescriptor) * this->m_MaxMessages;
		this->m_ConnDescBuf	= (MassiveConnectionDevice::ConnectionDescriptor*)pointer;
		pointer += sizeof(MassiveConnectionDevice::ConnectionDescriptor) * this->m_MaxConnections;
		this->m_ConnDescUnused = (MassiveConnectionDevice::ConnectionDescriptor**)pointer;
		pointer += sizeof(MassiveConnectionDevice::ConnectionDescriptor*) * this->m_MaxConnections;
		this->m_SendConnListBuf = (MassiveConnectionDevice::ConnectionDescriptor**)pointer;
		pointer += sizeof(MassiveConnectionDevice::ConnectionDescriptor*) * this->m_MaxConnections;
		this->m_RecvConnListBuf = (MassiveConnectionDevice::ConnectionDescriptor**)pointer;
		pointer += sizeof(MassiveConnectionDevice::ConnectionDescriptor*) * this->m_MaxConnections;
		this->m_SelectBuffer = (SOCKET*)pointer;
		pointer += sizeof(unsigned int) + sizeof(SOCKET) * this->m_MaxConnections;
		this->m_SendBuffers = (void**)pointer;
		pointer += sizeof(void*) * this->m_MaxConnections;
		this->m_RecvBuffers = (void**)pointer;
		pointer += sizeof(void*) * this->m_MaxConnections;
			
		for (unsigned long index = 0; index < this->m_MaxConnections; index++)
		{
			this->m_SendBuffers[index] = pointer;
			pointer += this->m_SendBufLength + sizeof(Internal::PackageHeader);
		}

		for (unsigned long index = 0; index < this->m_MaxConnections; index++)
		{
			this->m_RecvBuffers[index] = pointer;
			pointer += this->m_RecvBufLength + sizeof(Internal::PackageHeader);
		}


		// Building finish! Fill tables...
		for (unsigned long index = 0; index < this->m_MaxConnections; index++)
		{
			this->m_ConnDescBuf[index].sock				= INVALID_SOCKET;
			this->m_ConnDescBuf[index].group			= 0;
			this->m_ConnDescBuf[index].node				= 0;
			this->m_ConnDescBuf[index].key				= NULL;
			this->m_ConnDescBuf[index].remote_addr.s_addr = 0;
			this->m_ConnDescBuf[index].send_ptr			= (char*)this->m_SendBuffers[index];
			this->m_ConnDescBuf[index].recv_ptr			= (char*)this->m_RecvBuffers[index];
			this->m_ConnDescBuf[index].proc_ptr			= this->m_ConnDescBuf[index].recv_ptr;
			this->m_ConnDescBuf[index].send_left		= this->m_SendBufLength;
			this->m_ConnDescBuf[index].recv_left		= this->m_RecvBufLength;
			this->m_ConnDescBuf[index].bytes_sent		= 0;
			this->m_ConnDescBuf[index].bytes_recv		= 0;
			this->m_ConnDescBuf[index].is_listen		= 0;
			this->m_ConnDescBuf[index].is_recvblocked	= 0;
			this->m_ConnDescBuf[index].overflowed_count = 0;
			this->m_ConnDescBuf[index].flux_in_bytes	= 0;
			this->m_ConnDescBuf[index].flux_out_bytes	= 0;
			this->m_ConnDescBuf[index].flux_in_packets	= 0;
			this->m_ConnDescBuf[index].flux_out_packets	= 0;
			this->m_ConnDescBuf[index].flux_active_time = 0;
		}

		for (unsigned long index = 0; index < this->m_MaxConnections; index++)
		{
			this->m_ConnDescUnused[index] = &(this->m_ConnDescBuf[index]);
		}

		for (unsigned long index = 0; index < this->m_MaxMessages; index++)
		{
			this->m_MsgDescBuf[index].conn.group	= 0;
			this->m_MsgDescBuf[index].conn.node		= 0;
			this->m_MsgDescBuf[index].conn.routine	= 0;	//!!!!!
			this->m_MsgDescBuf[index].msge.time		= 0;	//!!!!!
			this->m_MsgDescBuf[index].msge.cmd		= 0;
			this->m_MsgDescBuf[index].msge.seg		= 0;
			this->m_MsgDescBuf[index].msge.len		= 0;
			this->m_MsgDescBuf[index].msge.buf		= NULL;
			this->m_MsgDescBuf[index].key			= NULL;
		}

		// Init Flux Info.
		this->m_FluxActiveTime	= 0;	//!!!!!!!!!
		this->m_FluxInBytes		= 0;
		this->m_FluxOutBytes	= 0;
		this->m_FluxInPackets	= 0;
		this->m_FluxOutPackets	= 0;
		this->m_FluxActiveTime	= 0;

		// All OK!
		DBGOUT("MassiveConnectionDevice Build Successful.");
		return true;
	}
	catch(const std::bad_alloc&)
	{
		Destroy();
	}
	
	return false;
}

void MassiveConnectionDevice::Destroy()
{
	if (this->m_ConnDescBuf != NULL)
	{
		for (unsigned long index = 0; index < this->m_MaxConnections; index++)
		{
			if (this->m_ConnDescBuf[index].sock != INVALID_SOCKET)
			{
				this->m_ConnDescBuf[index].is_recvblocked = true;
			}
		}

		// Flush Buffer.
		ConnEntity		conn;
		MessageEntity	msge;		
		this->RecvNetMessage(conn, msge, NULL);

		for (unsigned long index = 0; index < this->m_MaxConnections; index++)
		{
			if (this->m_ConnDescBuf[index].sock != INVALID_SOCKET)
			{
				closesocket(this->m_ConnDescBuf[index].sock);
			}
		}
	}

	this->m_MaxConnections	= 0;
	this->m_MaxMessages		= 0;
	this->m_SendBufLength	= 0;
	this->m_RecvBufLength	= 0;

	this->m_MsgDescCursor	= 0;
	this->m_MsgProcCursor	= 0;
	this->m_MsgCursor		= 0;
	this->m_ConnDescUnusedCursor = 0;
	this->m_SendConnCursor	= 0;
	this->m_RecvConnCursor	= 0;

	this->m_MsgDescBuf		= NULL;
	this->m_ConnDescBuf		= NULL;
	this->m_ConnDescUnused	= NULL;
	this->m_SendConnListBuf = NULL;
	this->m_SelectBuffer	= NULL;
	this->m_SendBuffers		= NULL;
	this->m_RecvBuffers		= NULL;

	this->m_FluxActiveTime	= 0;	//!!!!!!!!!
	this->m_FluxInBytes		= 0;
	this->m_FluxOutBytes	= 0;
	this->m_FluxInPackets	= 0;
	this->m_FluxOutPackets	= 0;
	this->m_FluxActiveTime	= 0;
	
	for (unsigned int i = 0; i < (sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0])); i++)
	{
		if (this->m_Node2ConnMapper[i] != NULL)
		{
			delete this->m_Node2ConnMapper[i];
			this->m_Node2ConnMapper[i] = NULL;
		}
	}

	if (this->m_Sock2ConnMapper != NULL)
	{
		delete this->m_Sock2ConnMapper;
		this->m_Sock2ConnMapper = NULL;
	}

	if (this->m_Sock2SendListMapper != NULL)
	{
		delete this->m_Sock2SendListMapper;
		this->m_Sock2SendListMapper = NULL;
	}

	if (this->m_Sock2RecvListMapper != NULL)
	{
		delete this->m_Sock2SendListMapper;
		this->m_Sock2RecvListMapper = NULL;
	}

	if (this->m_BasePointer != NULL)
	{
		delete[] this->m_BasePointer;
		this->m_BasePointer = NULL;
		DBGOUT(TEXT("MassiveConnectionDevice cleaned."));
	}
	return;
}

bool MassiveConnectionDevice::CreateNode(unsigned short group_id, unsigned long node_id, const char * local_ip, unsigned short listen_port, void * key)
{
	assert(local_ip != NULL);

	if ((!ExistNode(group_id, node_id)) && (this->m_ConnDescUnusedCursor > 0))
	{
		/* listen */
		SOCKET sock = socket(AF_INET, SOCK_STREAM, 0);
		if (sock != INVALID_SOCKET)
		{
			//===================== SET SOCKET OPTION =============================
			SetSocketReuseAddr(sock, true);		//debug mode
			SetSocketNoDelay(sock, true);
			SetSocketLinger(sock, false, 0);
			//=====================================================================

			sockaddr_in local_addr;
			local_addr.sin_family = PF_INET;
			local_addr.sin_addr.s_addr = inet_addr(local_ip);
			local_addr.sin_port = htons(listen_port);
			if (0 == bind(sock, (const sockaddr*)&local_addr, sizeof(local_addr)))
			{
				if (0 == listen(sock, 5))
				{
					LOG("Succeed create node @Node(%d:%lu) with addr:%s:%d.", group_id, node_id, local_ip, listen_port);
					return _InsertNode(group_id, node_id, sock, local_addr.sin_addr, key, true);
				}
			}
		}
	}	
	return false;
}

bool MassiveConnectionDevice::AppendNode(unsigned short group_id, unsigned long node_id, const char * node_ip, unsigned short port, void * key)
{
	assert(node_ip != NULL);

	LOG(TEXT("A new node prepare to connect @Node(%d:%lu), target addr:%s:%d"), group_id, node_id, node_ip, port);

	if ((!ExistNode(group_id, node_id)) && (this->m_ConnDescUnusedCursor > 0))
	{
		/* connect */
		SOCKET sock =  socket(AF_INET, SOCK_DGRAM, 0);
		if (sock != INVALID_SOCKET)
		{
			sockaddr_in remote_addr;
			remote_addr.sin_family = PF_INET;
			remote_addr.sin_addr.s_addr = inet_addr(node_ip);
			remote_addr.sin_port = htons(port);
			if (0 == connect(sock, (const sockaddr *)&remote_addr, sizeof(remote_addr)))
			{
				LOG("Succeed connect Node(%d:%lu) with addr:%s:%d.", group_id, node_id, node_ip, port);
				return _InsertNode(group_id, node_id, sock, remote_addr.sin_addr, key, false);
			}
		}
	}
	LOG("Failed to connect Node(%d:%lu) with addr:%s:%d.", group_id, node_id, node_ip, port);
	return false;
}

void MassiveConnectionDevice::RemoveNode(unsigned short group_id, unsigned long node_id, bool broken)
{
	if (this->m_Node2ConnMapper != NULL)
	{
		MassiveConnectionDevice::ConnectionDescriptor * conndesc = NULL;
		if (this->m_Node2ConnMapper[group_id %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->getlink(node_id, conndesc)
		&& (NULL != conndesc))
		{	
			assert(conndesc->sock != INVALID_SOCKET);

			//remove it from mapper.
			this->m_Node2ConnMapper[group_id %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->remove(node_id);

			LOG("Node(%d:%lu) removed.(%s)", group_id, node_id, broken? "Broken":"Destroyed");

			if (!conndesc->is_listen)
			{
				// Generate a disc-message.
			
				MessageEntity	tmpMsge;				
				if (broken)
				{
					tmpMsge.cmd = DEV_CONNECTION_BROKEN;
				}
				else
				{
					tmpMsge.cmd = DEV_CONNECTION_DESTROY;					
				}
				tmpMsge.buf		= NULL;
				tmpMsge.len		= 0;
				tmpMsge.time	= 0;	//time!!!

				
				ConnEntity		tmpConn;
				tmpConn.group	= group_id;
				tmpConn.node	= node_id;
				tmpConn.routine = 0;
			
				_RaiseMessage(tmpConn, tmpMsge, conndesc->key);
			}
	
		
			// remove from recv socket list.
			MassiveConnectionDevice::ConnectionDescriptor ** lpconndesc = NULL;
			if ((this->m_Sock2RecvListMapper->getlink((unsigned long)(conndesc->sock), lpconndesc))
			&& (NULL != lpconndesc))			
			{				
				this->m_RecvConnCursor--;
				*lpconndesc = this->m_RecvConnListBuf[this->m_RecvConnCursor]; 
				
				assert((*lpconndesc)->sock != INVALID_SOCKET);

				this->m_Sock2RecvListMapper->remove((unsigned long)(conndesc->sock));
				this->m_Sock2RecvListMapper->replace((unsigned long)((*lpconndesc)->sock), lpconndesc);
			}

			// clean this connection.
			if (conndesc->sock != INVALID_SOCKET)
			{
				this->m_Sock2ConnMapper->remove((unsigned long)(conndesc->sock));
				closesocket(conndesc->sock);
			}

			//init connection information
			conndesc->sock				= INVALID_SOCKET;
			conndesc->group				= 0;
			conndesc->node				= 0;
			conndesc->key				= NULL;
			conndesc->bytes_recv		= 0;
			conndesc->bytes_sent		= 0;
			conndesc->recv_ptr = (char*)this->m_RecvBuffers[(unsigned long)(((char*)conndesc - (char*)(this->m_ConnDescBuf))/sizeof(ConnectionDescriptor))];
			conndesc->send_ptr = (char*)this->m_SendBuffers[(unsigned long)(((char*)conndesc - (char*)(this->m_ConnDescBuf))/sizeof(ConnectionDescriptor))];
			conndesc->recv_left			= this->m_RecvBufLength;
			conndesc->send_left			= this->m_SendBufLength;
			conndesc->proc_ptr			= conndesc->recv_ptr;
			conndesc->is_listen			= 0;
			conndesc->is_recvblocked	= 0;
			conndesc->overflowed_count	= 0;
			conndesc->flux_in_bytes		= 0;
			conndesc->flux_in_packets	= 0;
			conndesc->flux_out_bytes	= 0;
			conndesc->flux_out_packets	= 0;
			conndesc->flux_active_time	= 0;

			//push to unused connection descriptor stack.
			this->m_ConnDescUnused[this->m_ConnDescUnusedCursor++] = conndesc;
			
			LOG("%lu/%lu connection(s).", this->m_MaxConnections - this->m_ConnDescUnusedCursor, this->m_MaxConnections);

		}
	}
	return;
}

bool MassiveConnectionDevice::ExistNode(unsigned short group_id, unsigned long node_id)
{
	MassiveConnectionDevice::ConnectionDescriptor * temp = NULL;
	return ((this->m_Node2ConnMapper[group_id %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))] != NULL) 
		&& this->m_Node2ConnMapper[group_id %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->getlink((unsigned long)(node_id), temp));
}


bool MassiveConnectionDevice::MoveNode(unsigned short group_id, unsigned long node_id, unsigned short new_group_id, unsigned long new_node_id)
{
	MassiveConnectionDevice::ConnectionDescriptor * point = NULL;
	if (this->m_Node2ConnMapper[group_id %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->getlink((unsigned long)(node_id), point)
	&& (point != NULL))
	{		
		if (this->m_Node2ConnMapper[new_group_id %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->insert(new_node_id, point))
		{
			point->group = new_group_id;
			point->node = new_node_id;	
			this->m_Node2ConnMapper[group_id %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->remove(node_id);

			LOG("Node(%d:%lu) Move to: (%d:%lu).\n", group_id, node_id, new_group_id, new_node_id);

			return true;
		}
	}
	WARNING("Failed to Move Node(%d:%lu) to (%d:%lu).\n", group_id, node_id, new_group_id, new_node_id);
	return false;
}

bool MassiveConnectionDevice::LockNode(unsigned short group_id, unsigned long node_id, bool locked)
{
	MassiveConnectionDevice::ConnectionDescriptor * point = NULL;
	if (this->m_Node2ConnMapper[group_id %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->getlink((unsigned long)(node_id), point)
	&& (point != NULL))
	{
		point->is_recvblocked = locked;
		DBGOUT("Node(%d:%lu) %s.", group_id, node_id, (locked ? TEXT("Locked"):TEXT("Unlocked")));
		return true;
	}
	return false;
}

unsigned long MassiveConnectionDevice::GetNodeCount(unsigned short group)
{
	return this->m_Node2ConnMapper[group %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->count();
}

bool MassiveConnectionDevice::BindNodeKey(unsigned short group_id, unsigned long node_id, void * key)
{
	MassiveConnectionDevice::ConnectionDescriptor * point = NULL;
	if (this->m_Node2ConnMapper[group_id %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->getlink((unsigned long)(node_id), point)
	&& (point != NULL))
	{
		point->key = key;
		DBGOUT("Node(%d:%lu) Key Changed.", group_id, node_id);
		return true;
	}
	return false;
}

bool MassiveConnectionDevice::GetFluxInfo(unsigned short group, unsigned long node, FluxInfo * flux)
{
	assert(flux != NULL);

	if (flux != NULL)
	{
		MassiveConnectionDevice::ConnectionDescriptor * conndesc = NULL;
		if (this->m_Node2ConnMapper[group %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->getlink((unsigned long)(node), conndesc)
		&& (conndesc != NULL))
		{
			flux->in_bytes		= conndesc->flux_in_bytes;
			flux->out_bytes		= conndesc->flux_out_bytes;
			flux->in_packets	= conndesc->flux_in_packets;
			flux->out_packets	= conndesc->flux_out_packets;
			flux->active_time	= conndesc->flux_active_time;
			flux->capture_time	= 0;	//!!!!!!!!
			return true;
		}
	}
	return false;
}

bool MassiveConnectionDevice::GetFluxInfo(FluxInfo * flux)
{
	assert(flux != NULL);

	if (flux != NULL)
	{
		flux->in_bytes		= this->m_FluxInBytes;
		flux->out_bytes		= this->m_FluxOutBytes;
		flux->in_packets	= this->m_FluxInPackets;
		flux->out_packets	= this->m_FluxOutPackets;
		flux->active_time	= this->m_FluxActiveTime;
		flux->capture_time	= 0;	//!!!!!!
		return true;
	}
	return false;
}

unsigned long MassiveConnectionDevice::GetNodeIP(unsigned short group, unsigned long node)
{
	MassiveConnectionDevice::ConnectionDescriptor * conndesc = NULL;
	if (this->m_Node2ConnMapper[group %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->getlink((unsigned long)(node), conndesc)
	&& (conndesc != NULL))
	{
		return conndesc->remote_addr.s_addr;
	}
	return 0;
}


bool MassiveConnectionDevice::SendNetMessage(const ConnEntity& conn, const MessageEntity & msge, AppendBuffer * append, unsigned long append_count)
{
	assert(!(msge.len !=0 && msge.buf == NULL));

	if (this->m_Node2ConnMapper != NULL)
	{
		DBGOUT("A message send. cmd:%d, seg:%d, len:%lu, time:%lu, buf:%8s\n", 
			msge.cmd, msge.seg, msge.len, msge.time, (msge.buf == NULL ? TEXT("NULL"):(const char *)(msge.buf)));

		MassiveConnectionDevice::ConnectionDescriptor * point = NULL;
		if (this->m_Node2ConnMapper[conn.group %(sizeof(this->m_Node2ConnMapper)/sizeof(this->m_Node2ConnMapper[0]))]->getlink((unsigned long)(conn.node), point) && point != NULL)
		{
			unsigned long append_bytes = 0;
			if ((append != NULL) && (append_count != 0))
			{
				for (unsigned long i = 0; i < append_count; i++)
				{
					append_bytes += append[i].len;
				}
			}

			if (point->send_left >=  sizeof(Internal::PackageHeader) + msge.len + append_bytes)
			{	
				if (point->send_left == this->m_SendBufLength)
				{
					this->m_SendConnListBuf[this->m_SendConnCursor] = point;
					this->m_Sock2SendListMapper->replace((unsigned long)(point->sock), &(this->m_SendConnListBuf[this->m_SendConnCursor]));
					this->m_SendConnCursor++;
				}
				
				Internal::PackageHeader header;
				header.length = (unsigned short)((msge.len + append_bytes) < 65536 ? (msge.len + append_bytes) : 65536);
				header.routine = conn.routine;
				header.command = msge.cmd;
				header.segment = msge.seg;
				
				memcpy(point->send_ptr, &header, sizeof(Internal::PackageHeader));
				if (msge.buf != NULL) memcpy(point->send_ptr + sizeof(Internal::PackageHeader), msge.buf, msge.len);
				point->send_ptr	+= sizeof(Internal::PackageHeader) + msge.len;
				for (unsigned long i = 0; i < append_count; i++)
				{
					memcpy(point->send_ptr, append[i].buf, append[i].len);
					point->send_ptr += append[i].len;
				}
				point->send_left -= sizeof(Internal::PackageHeader)+ msge.len + append_bytes;

				//------------------------------
				//save the flux_in information.
				point->flux_out_packets++;
				this->m_FluxOutPackets++;
				//------------------------------
				return true;
			}
			else
			{
				//Overflowed !!!
				assert(!"Buffer overlow!");
				RemoveNode(conn.group, conn.node, true);
			}
		}
	}
	return false;
}

bool MassiveConnectionDevice::RecvNetMessage(ConnEntity & conn, MessageEntity & msge, void ** lpkey)
{
	/*
		1st. Look up the message queue, if any message there, pop it, and return.
	*/
	if (this->m_MsgDescCursor > this->m_MsgProcCursor)
	{
		conn = this->m_MsgDescBuf[m_MsgProcCursor].conn;
		msge = this->m_MsgDescBuf[m_MsgProcCursor].msge;
		if (lpkey != NULL) *lpkey = this->m_MsgDescBuf[m_MsgProcCursor].key;
		if (++this->m_MsgProcCursor == this->m_MsgDescCursor)
		{
			this->m_MsgDescCursor = this->m_MsgProcCursor = 0;
		}
		return true;
	}

	/*
		2nd. Send all data.
	*/
	if (this->m_SendConnCursor > 0)
	{
#if defined (_WIN32)
		int count = 0;
		for (; count < (int)(this->m_SendConnCursor); count++)
		{
			this->m_SelectBuffer[count + 1] = this->m_SendConnListBuf[count]->sock;
		}
		this->m_SelectBuffer[0] = this->m_SendConnCursor;
		count = select(count, NULL, (fd_set*)this->m_SelectBuffer, NULL, &(this->m_SendTimeout));
#else
		SOCKET maxfd = 0;
		fd_set writefd;
		FD_ZERO(&writefd);		
		for (unsigned long i = 0; i < this->m_SendConnCursor; i++)
		{			
			if (this->m_SendConnListBuf[i]->sock > maxfd)
			{
				maxfd = this->m_SendConnListBuf[i]->sock;
			}			
			FD_SET(this->m_SendConnListBuf[i]->sock, &writefd);
		}
		int count = select((int)maxfd + 1, NULL, &writefd, NULL, &(this->m_SendTimeout));
#endif
		if (count != 0 && count != SOCKET_ERROR)
		{
			MassiveConnectionDevice::ConnectionDescriptor ** lpconndesc = NULL;
			MassiveConnectionDevice::ConnectionDescriptor *  conndesc = NULL;
#if defined(_WIN32)
			for (int i = 1; i <= count; i++)
			{
				if ((this->m_Sock2SendListMapper->getlink((unsigned long)(m_SelectBuffer[i]), lpconndesc))
				&& (NULL != lpconndesc)
				&& (NULL != (conndesc = *lpconndesc)))
#else
			for (int i = 0; i < count; i++)
			{
				if ((FD_ISSET(this->m_SendConnListBuf[i]->sock, &writefd))
				&& (this->m_Sock2SendListMapper->getlink((unsigned long)(this->m_SendConnListBuf[i]->sock), lpconndesc))
				&& (NULL != lpconndesc)
				&& (NULL != (conndesc = *lpconndesc)))

#endif
				{
					char * mark = conndesc->send_ptr + conndesc->send_left + conndesc->bytes_sent - this->m_SendBufLength;
					int sent = send(conndesc->sock, mark, int(conndesc->send_ptr - mark), 0);
					if (sent != SOCKET_ERROR)
					{
						conndesc->bytes_sent += sent;
						if (conndesc->bytes_sent == this->m_SendBufLength - conndesc->send_left)
						{
							//------------------------------
							//save the flux_out information.
							conndesc->flux_out_bytes += conndesc->bytes_sent;
							this->m_FluxOutBytes += conndesc->bytes_sent;
							//------------------------------

							// reset
							conndesc->bytes_sent = 0;
							conndesc->send_ptr = conndesc->send_ptr + conndesc->send_left - this->m_SendBufLength;
							conndesc->send_left = this->m_SendBufLength;


							this->m_SendConnCursor--;
							assert(this->m_SendConnCursor != (unsigned long)(-1));
							*lpconndesc = this->m_SendConnListBuf[this->m_SendConnCursor];
							this->m_Sock2SendListMapper->remove((unsigned long)(conndesc->sock));
							this->m_Sock2SendListMapper->replace((unsigned long)((*lpconndesc)->sock), lpconndesc);
							
						}
					}
				}
			}
		}
	}

	/* 
		3rd. If message queue is empty, recv would waitting until any message arrived or timeout,
		then the first message returns and the other messages push to message queue.
		when recv a new connection, generate a new-conn-msg and insert into the connection list.
		when recv a disconnection, generate a disc-conn-msg and remove from the connection list.
		when recv a shutdown message, no message returned but false returned.(Method Destroy() could cause this message.)
	*/

	// Recv Messages.
	if (this->m_RecvConnCursor > 0)
	{
#if defined(_WIN32)	
		int count = 0;
		for (; count < (int)(this->m_RecvConnCursor); count++)
		{
			this->m_SelectBuffer[count + 1] = this->m_RecvConnListBuf[count]->sock;
		}
		this->m_SelectBuffer[0] = this->m_RecvConnCursor;
		count = select(count, (fd_set*)this->m_SelectBuffer, NULL, NULL, &(this->m_RecvTimeout));
#else
		SOCKET maxfd = 0;
		fd_set readfd;
		FD_ZERO(&readfd);
		for (unsigned long i = 0; i < this->m_RecvConnCursor; i++)
		{
			if (this->m_RecvConnListBuf[i]->sock > maxfd)
			{
				maxfd = this->m_RecvConnListBuf[i]->sock;
			}		
			FD_SET(this->m_RecvConnListBuf[i]->sock, &readfd);
		}
		int count = select((int)maxfd + 1, &readfd, NULL, NULL, &(this->m_RecvTimeout));
#endif		
		if (count != 0 && count != SOCKET_ERROR)
		{
			ConnectionDescriptor *  conndesc = NULL;
#if defined(_WIN32)
			for (int i = 1; i <= count; i++)
			{			
				if (this->m_Sock2ConnMapper->getlink((unsigned long)(this->m_SelectBuffer[i]), conndesc)
				&& (conndesc != NULL))
#else			
			
			for (int i = 0; i < int(this->m_RecvConnCursor); i++)
			{						
				if ((FD_ISSET(this->m_RecvConnListBuf[i]->sock, &readfd))
				&& (this->m_Sock2ConnMapper->getlink((unsigned long)(this->m_RecvConnListBuf[i]->sock), conndesc))
				&& (conndesc != NULL))
#endif
				{
					if (conndesc->is_listen)
					{
						sockaddr_in address;
						socklen_t	len = sizeof(address);

						//Is listen socket
						SOCKET sock = accept(conndesc->sock, (sockaddr*)&address, &len);
						if (sock != INVALID_SOCKET)
						{						  
							// Recv a new connection
							//+++++++++++++++++++++++++++++++++++++++++++++
							ConnEntity tmpConn;
					
							tmpConn.group	= 0;
							tmpConn.node	= (unsigned long)sock;
							tmpConn.routine = 0;

							if (!_InsertNode(tmpConn.group, tmpConn.node, sock, address.sin_addr, conndesc->key, false))	// a conn-create-message generated.
							{
								closesocket(sock);
							}
							//+++++++++++++++++++++++++++++++++++++++++++++
						}
					}
					else if (!conndesc->is_recvblocked)
					{
						//Is normal socket
						int received = recv(conndesc->sock, conndesc->recv_ptr, conndesc->recv_left, 0);
						if (received != SOCKET_ERROR && received != 0)
						{
							conndesc->bytes_recv += received;
							conndesc->recv_ptr += received;
							conndesc->recv_left -= received;
							while (conndesc->bytes_recv >= sizeof(Internal::PackageHeader))
							{
								Internal::PackageHeader * header = (Internal::PackageHeader*)conndesc->proc_ptr;
																
								// Is a data packet
								if (conndesc->bytes_recv < header->length)
								{
									if (conndesc->recv_left < header->length - conndesc->bytes_recv)
									{
										conndesc->recv_left = header->length - conndesc->bytes_recv;
									}
									break;
								}
								else
								{
									//*************************************************************************
									// construct message.
									MassiveConnectionDevice::MessageDescriptor * msgdesc = &(this->m_MsgDescBuf[this->m_MsgDescCursor++]);

									msgdesc->conn.group		= conndesc->group;
									msgdesc->conn.node		= conndesc->node;
									msgdesc->conn.routine	= header->routine;
									msgdesc->msge.cmd		= header->command;
									msgdesc->msge.seg		= header->segment;
									msgdesc->msge.buf		= (char *)(header) + sizeof(Internal::PackageHeader);
									msgdesc->msge.len		= header->length;
									msgdesc->msge.time		= 0;				//!!!
									msgdesc->key			= conndesc->key;

									//------------------------------
									//save the flux information
									conndesc->flux_in_packets++;
									this->m_FluxInPackets++;
									//------------------------------


									LOG(TEXT("A message received from Node(%d:%lu[%d]::%d), cmd:%d, seg:%d, len:%lu, time:%lu"),
										msgdesc->conn.group, msgdesc->conn.node, msgdesc->conn.routine, conndesc->sock,
										msgdesc->msge.cmd, msgdesc->msge.seg,
										msgdesc->msge.len, msgdesc->msge.time);

									//*************************************************************************
									conndesc->bytes_recv -= sizeof(Internal::PackageHeader) + header->length;
									if (conndesc->bytes_recv == 0)
									{
										// reset stream
										conndesc->recv_ptr = (char*)this->m_RecvBuffers[(unsigned long)(((char*)conndesc - (char*)(this->m_ConnDescBuf))/sizeof(ConnectionDescriptor))];

										//------------------------------
										//save the flux_in information.
										conndesc->flux_in_bytes += conndesc->proc_ptr - conndesc->recv_ptr;
										this->m_FluxInBytes += conndesc->proc_ptr - conndesc->recv_ptr;
										//------------------------------

										conndesc->proc_ptr = conndesc->recv_ptr;
										conndesc->recv_left = this->m_RecvBufLength;
										break;
									}
									conndesc->proc_ptr += sizeof(Internal::PackageHeader) + header->length;
								}
							}
						}
						else
						{
							// connection breaked or error occured
							//++++++++++++++++++++++++++++++++++++++++++++

							RemoveNode(conndesc->group, conndesc->node, true);

							//++++++++++++++++++++++++++++++++++++++++++++
						}
					}//recv finish
				}
			}// Loop to process each socket.
		}

		// pop a message.
		if (this->m_MsgDescCursor > this->m_MsgProcCursor)
		{
			conn = this->m_MsgDescBuf[m_MsgProcCursor].conn;
			msge = this->m_MsgDescBuf[m_MsgProcCursor].msge;
			if (lpkey != NULL) *lpkey = this->m_MsgDescBuf[m_MsgProcCursor].key;
			if (++this->m_MsgProcCursor == this->m_MsgDescCursor)
			{
				this->m_MsgDescCursor = this->m_MsgProcCursor = 0;
			}
			return true;
		}		
	}
	return false;
}
